package com.allen.study.test_common.utils.kafka;

import com.allen.study.test_common.exception.CustomRuntimeException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @ClassName: KafkaUtils
 * @Author: AllenSun
 * @Date: 2025/3/5 17:13
 */
@Component
@Slf4j
public class KafkaUtils {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 发送消息到指定主题
     *
     * @param topic   主题名称
     * @param message 消息内容
     */
    public void sendMessage(String topic, String message) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                throw new CustomRuntimeException("kafka消息发送失败：{}"+ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("kafka消息发送成功: " + message + ", partition: " + result.getRecordMetadata().partition() + ", " +
                        "offset: " + result.getRecordMetadata().offset());
            }
        });
    }
}
